package com.zx._02_thread.线程池;

import java.util.ArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池的oom：https://mp.weixin.qq.com/s/-ff7Enr4AMUDi8MR1LQqzw
 */
public class 线程池bug_内存泄漏 {

    public static void main(String[] args) throws Exception {

        ThreadPoolExecutor productThreadPoolExecutor = new ThreadPoolExecutor(1,
                1,
                1,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1),
                new MyThreadFactory("product"),
                new MyRejectedPolicy() // 使用非抛出异常的策略，导致future一致阻塞，最终oom
                //new AbortPolicy()   //使用默认策略，不会导致oom
        );

        while (true) {
            TimeUnit.MILLISECONDS.sleep(50L);
            new Thread(() -> {
                ArrayList<Future<Integer>> futureList = new ArrayList<>();
                //从数据库获取产品信息
                int productNum = 5;
                for (int i = 0; i < productNum; i++) {
                    try {
                        int finalI = i;
                        Future<Integer> future = productThreadPoolExecutor.submit(() -> {
                            System.out.println(
                                    "Thread.currentThread().getName() = " + Thread.currentThread().getName());
                            return finalI * 10;
                        });
                        futureList.add(future);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                for (Future<Integer> integerFuture : futureList) {
                    try {
                        // 线程池中的任务触发拒绝策略后，Future被丢弃，任务不会完成，就永远阻塞在这里。
                        // 该 bug 在jdk9的时候被修复。
                        // 或者使用默认的拒绝策略，抛出异常，导致future.get 发送中断。
                        Integer integer = integerFuture.get();
                        System.out.println(integer);
                        System.out.println("future.get() = " + integer);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }

    }

    static class MyThreadFactory implements ThreadFactory {

        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;
        private final String threadFactoryName;

        MyThreadFactory(String threadStartName) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                    Thread.currentThread().getThreadGroup();
            namePrefix = threadStartName + "-pool-" +
                    poolNumber.getAndIncrement() +
                    "-thread-";
            threadFactoryName = threadStartName;
        }

        public String getThreadFactoryName() {
            return threadFactoryName;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                    namePrefix + threadNumber.getAndIncrement(),
                    0);
            if (t.isDaemon()) {
                t.setDaemon(false);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    public static class MyRejectedPolicy implements RejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (e.getThreadFactory() instanceof MyThreadFactory) {
                MyThreadFactory myThreadFactory = (MyThreadFactory) e.getThreadFactory();
                if ("product".equals(myThreadFactory.getThreadFactoryName())) {
                    System.out.println("线程池有任务被拒绝了,请关注");
                }
            }
        }
    }
}
